package com.ecloud.hbase.service.impl;

import com.ecloud.common.ResultEnum;
import com.ecloud.common.exception.ServerException;
import com.ecloud.hbase.config.HbaseConnFactory;
import com.ecloud.hbase.service.HbaseService;
import com.ecloud.hbase.standard.GetInner;
import com.ecloud.hbase.standard.HbaseTable;
import com.ecloud.hbase.util.HbaseTemplate;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.lang.reflect.Field;

/**
 * hbase operator
 *
 * @author Squbi
 * @date 2021/6/16 11:38
 */
@Service
public class HbaseServiceImpl implements HbaseService {
    private static final Logger logger = LoggerFactory.getLogger(HbaseServiceImpl.class);

    @Autowired
    HbaseConnFactory factory;

    @Autowired
    HbaseTemplate template;

    @Override
    public boolean isTableExist(String tableName) {
        return isTableExist(TableName.valueOf(tableName));
    }

    @Override
    public boolean isTableExist(TableName tableName) {
        try {
            return factory.getAdmin().isTableAvailable(tableName);
        } catch (IOException e) {
            e.printStackTrace();
            throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
        }
    }

    @Override
    public <T extends HbaseTable> boolean isTableExist(Class<T> tableClass) {
        return isTableExist(tableClass.getSimpleName());
    }

    @Override
    public <T extends HbaseTable> void createTable(Class<T> tableClass) {
        String tableName = tableClass.getSimpleName();
        //判断表是否存在
        if (!isTableExist(tableName)) {
            //创建表属性对象,表名需要转字节
            HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
            //创建多个列族
            for (Field field : tableClass.getDeclaredFields()) {
                descriptor.addFamily(new HColumnDescriptor(field.getName()));
            }
            try {
                //根据对表的配置，创建表
                factory.getAdmin().createTable(descriptor);
                logger.debug("create table {} success", tableName);
            } catch (Exception e) {
                e.printStackTrace();
                throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
            }
        }
    }

    @Override
    public <T extends HbaseTable> void setData(T table, String rowKey) {
        // 如果表不存在 创建表
        Class<? extends HbaseTable> tableClass = table.getClass();
        if (!isTableExist(tableClass)) {
            createTable(tableClass);
        }
        // 反射将类字段映射到表
        String tableName = tableClass.getSimpleName();
        template.put(tableName, rowKey, put -> {
            Put p = table.toPut(put);
            logger.debug(" -- create put to push {}", p);
            return p;
        });
    }

    @Override
    public <T extends HbaseTable> T getData(Class<T> tableClass, String rowKey) {
        return getData(tableClass, rowKey, get -> get);
    }

    @Override
    public <T extends HbaseTable> T getData(Class<T> tableClass, String rowKey, GetInner inner) {
        Result result = getResult(tableClass, rowKey, inner);
        return parseResult(result, tableClass);
    }

    @Override
    public <T extends HbaseTable> Result getResult(Class<T> tableClass, String rowKey, GetInner inner) {
        String tableName = tableClass.getSimpleName();
        if (!isTableExist(tableName)) {
            throw new ServerException(ResultEnum.REQUEST_NULL);
        }
        return template.get(tableName, rowKey, get -> {
            for (Field field : tableClass.getDeclaredFields()) {
                get.addFamily(Bytes.toBytes(field.getName()));
            }
            return inner.getIn(get);
        });
    }

    @Override
    public <T extends HbaseTable> T getData(Class<T> tableClass, Get get) {
        String tableName = tableClass.getSimpleName();
        if (!isTableExist(tableName)) {
            throw new ServerException(ResultEnum.REQUEST_NULL);
        }
        Result result = template.get(tableName, get);
        logger.debug("get data from {} result : {}", tableName, result);
        return parseResult(result, tableClass);
    }

    @Override
    public <T extends HbaseTable> T parseResult(Result result, Class<T> tableClass) {
        if (result.isEmpty()) {
            return null;
        }
        try {
            T t = tableClass.getConstructor().newInstance();
            t.parseResult(result);
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            throw new ServerException(ResultEnum.SYS_ERROR, e.getMessage());
        }
    }

    @Override
    public void deleteTable(String tableNameStr) {
        Admin admin;
        try {
            admin = factory.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
            throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
        }

        if (isTableExist(tableNameStr)) {
            TableName tableName = TableName.valueOf(tableNameStr);
            boolean tableDisabled;
            try {
                tableDisabled = admin.isTableDisabled(tableName);
            } catch (IOException e) {
                e.printStackTrace();
                throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
            }

            if (!tableDisabled) {
                try {
                    admin.disableTable(tableName);
                } catch (IOException e) {
                    e.printStackTrace();
                    throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
                }
            }

            try {
                admin.deleteTable(tableName);
            } catch (IOException e) {
                e.printStackTrace();
                throw new ServerException(ResultEnum.DB_INNER_ERROR, e.getMessage());
            }
        } else {
            throw new ServerException(ResultEnum.REQUEST_NULL);
        }
    }
}
